package io.github.eric.urpc.server.handler;

import io.github.eric.urpc.core.rpc.Message;
import io.github.eric.urpc.core.thread.OrderedThreadPoolExecutor;
import io.github.eric.urpc.server.ServerContext;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

@Sharable
public class ServerStatHandler extends ServerBaseHandler {
    private static AtomicInteger requests = new AtomicInteger(0);
    private static AtomicLong latency = new AtomicLong(0);
    private static ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
    private static ScheduledExecutorService sysTimeExecutorService = Executors.newScheduledThreadPool(1);
    private static int PERIOD = 10;
    private static long cachedSystemTime;

    public ServerStatHandler(ServerBootstrap serverBootstrap, ServerContext serverContext) {
        super(serverBootstrap, serverContext);
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    stat();
                } catch (Exception e) {
                    serverContext.getLogger().error("[urpc server] stat error", e);
                }
            }
        }, PERIOD, PERIOD, TimeUnit.SECONDS);
        sysTimeExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    cachedSystemTime = System.currentTimeMillis();
                }
            }
        }, 0, 20, TimeUnit.MILLISECONDS);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        requests.addAndGet(1);
        Message message = (Message) msg;
        if (message != null) {
            latency.addAndGet(cachedSystemTime - message.getSendAt());
        }
    }

    public void stat() {
        int total = requests.getAndSet(0);
        int qps = total / PERIOD;
        int size = 0;
        if (serverContext.getRpcExecutor() instanceof ThreadPoolExecutor) {
            size = ((ThreadPoolExecutor) serverContext.getRpcExecutor()).getQueue().size();
        } else {
            size = ((OrderedThreadPoolExecutor) serverContext.getRpcExecutor()).getQueueSize();
        }
        long totalLatency = latency.getAndSet(0);
        long avgLatency = 0;
        if (total > 0) {
            avgLatency = totalLatency / total;
        }
        serverContext.getLogger().info("[urpc server] current QPS: {}, queue size: {}, avg latency: {}ms", qps, size, avgLatency);
    }
}
